#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
农业数据可视化Web平台 - 添加同步Hive功能版本
"""

import pandas as pd
import numpy as np
import json
import os
from datetime import datetime, timedelta
import logging
import matplotlib.pyplot as plt
import seaborn as sns
import warnings
import matplotlib.font_manager as fm
from matplotlib.font_manager import FontProperties
from flask import Flask, jsonify, request, render_template, Response
from flask_cors import CORS
import io

# 导入Hive连接相关库
try:
    from pyhive import hive
    HIVE_AVAILABLE = True
except ImportError:
    HIVE_AVAILABLE = False
    print("警告: pyhive未安装，将使用模拟数据")

warnings.filterwarnings('ignore')

class AgricultureDataService:
    def __init__(self):
        self.setup_logging()
        self.setup_product_mapping()
        self.hive_conn = None
        self.df = None
        self.base_dir = self.setup_directories()
        self.setup_chinese_fonts()
        self.connect_hive()
        self.load_data()

    def setup_product_mapping(self):
        """设置产品名称映射和模糊匹配"""
        # 标准产品名称列表
        self.standard_products = [
            '大白菜', '小白菜', '萝卜', '土豆', '番茄', '黄瓜', '茄子', '辣椒',
            '豆角', '苹果', '梨', '桃子', '葡萄', '香蕉', '西瓜', '胡萝卜',
            '菠菜', '韭菜', '芹菜', '生菜', '菜花', '豆芽', '蒜苗', '油菜', '茴香'
        ]
        
        # 产品别名映射字典
        self.product_aliases = {
            '萝卜': ['白萝卜', '青萝卜', '水萝卜', '樱桃萝卜', '红萝卜', '心里美萝卜'],
            '辣椒': ['青椒', '红椒', '尖椒', '甜椒', '干辣椒', '小辣椒', '朝天椒'],
            '大白菜': ['白菜', '娃娃菜', '包菜'],
            '小白菜': ['青菜',  '小青菜'],
            '番茄': ['西红柿', '圣女果', '小番茄'],
            '土豆': ['马铃薯', '洋芋', '地蛋'],
            '胡萝卜': ['红萝卜', '黄萝卜'],
            '豆角': ['四季豆', '长豆角', '扁豆', '豇豆'],
            '菜花': ['花菜', '花椰菜', '西兰花'],
            '生菜': ['球生菜', '叶生菜', '奶白菜'],
            '芹菜': ['西芹', '本芹', '水芹', '芹菜叶']
        }

    def fuzzy_match_product(self, product_name):
        """使用模糊匹配找到最相似的标准产品名称"""
        if not product_name:
            return None
            
        # 首先检查是否在别名映射中
        for standard_name, aliases in self.product_aliases.items():
            if product_name in aliases:
                return standard_name
        
        # 检查是否包含关键词
        for standard_name in self.standard_products:
            if standard_name in product_name or product_name in standard_name:
                return standard_name
        
        return product_name

    def setup_chinese_fonts(self):
        """设置中文字体"""
        try:
            font_candidates = [
                '/usr/local/lib/python3.6/site-packages/mplfonts/fonts/NotoSansCJKsc-Regular.otf',
                '/usr/share/fonts/wqy-zenhei/wqy-zenhei.ttc',
                '/usr/share/fonts/wqy-microhei/wqy-microhei.ttc'
            ]
            
            self.chinese_font = None
            self.font_path = None
            
            for font_path in font_candidates:
                if os.path.exists(font_path):
                    try:
                        self.chinese_font = FontProperties(fname=font_path)
                        self.font_path = font_path
                        print(f"✅ 成功设置字体: {font_path}")
                        break
                    except Exception as e:
                        continue
            
            self.use_chinese = self.chinese_font is not None
            
        except Exception as e:
            self.use_chinese = False

    def setup_directories(self):
        """设置目录"""
        current_dir = os.path.dirname(os.path.abspath(__file__))
        
        directories = {
            'base': current_dir,
            'results': os.path.join(current_dir, 'analysis_results'),
            'static': os.path.join(current_dir, 'static'),
            'charts': os.path.join(current_dir, 'static', 'charts'),
            'logs': os.path.join(current_dir, 'logs'),
            'templates': os.path.join(current_dir, 'templates')
        }
        
        for dir_name, dir_path in directories.items():
            os.makedirs(dir_path, exist_ok=True)
        
        return directories

    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.StreamHandler()
            ]
        )

    def connect_hive(self):
        """连接Hive数据库"""
        try:
            if not HIVE_AVAILABLE:
                logging.warning("PyHive未安装，无法连接Hive")
                return False
                
            # 连接到Hive
            self.hive_conn = hive.Connection(
                host='localhost',
                port=10000,
                database='default'
            )
            
            # 测试连接
            cursor = self.hive_conn.cursor()
            cursor.execute("SELECT 1")
            result = cursor.fetchone()
            cursor.close()
            
            logging.info("Hive数据库连接成功")
            return True
            
        except Exception as e:
            logging.error(f"Hive数据库连接失败: {e}")
            self.hive_conn = None
            return False

    def load_data(self):
        """加载数据"""
        try:
            if self.load_data_from_hive():
                return True
            else:
                logging.warning("从Hive加载数据失败，使用模拟数据")
                return self.load_sample_data()
        except Exception as e:
            logging.error(f"数据加载失败: {e}")
            return self.load_sample_data()

    def load_data_from_hive(self):
        """从Hive加载数据 - 使用真实表结构"""
        try:
            if not self.hive_conn:
                return False
            
            # 首先检查表是否存在以及数据量
            cursor = self.hive_conn.cursor()
            cursor.execute("SELECT 1 FROM agriculture_cleaned_data LIMIT 1")
            has_data = cursor.fetchone()
            cursor.close()
            
            if has_data:
                logging.info("检测到Hive表中有数据，开始加载...")
                
                # 根据真实表结构修改查询
                query = """
                SELECT 
                    record_id,
                    product_name,
                    product_category,
                    avg_price,
                    low_price,
                    high_price,
                    place,
                    spec,
                    unit,
                    pub_date,
                    crawl_time,
                    price_anomaly_flag,
                    data_quality_score,
                    processing_timestamp,
                    batch_id,
                    data_source
                FROM agriculture_cleaned_data
                ORDER BY pub_date DESC
                LIMIT 1000
                """
                
                self.df = pd.read_sql(query, self.hive_conn)
                
                # 数据清理和处理
                self.df = self.clean_hive_data_real(self.df)
                
                logging.info(f"从Hive加载数据成功，共 {len(self.df)} 条记录")
                return True
            else:
                logging.warning("Hive表中没有数据")
                return False
                
        except Exception as e:
            logging.error(f"从Hive加载数据失败: {e}")
            return False

def clean_hive_data_real(self, df):
        """清理从Hive加载的真实数据"""
        try:
            # 处理空值
            df = df.fillna({
                'place': '未知',
                'unit': '斤',
                'product_category': '蔬菜',
                'price_anomaly_flag': 'NORMAL',
                'data_quality_score': 90,
                'spec': '普通',
                'batch_id': '默认批次',
                'data_source': '市场数据'
            })
            
            # 创建代码需要的字段
            df['standard_product_name'] = df['product_name'].apply(self.fuzzy_match_product)
            df['volume'] = 1000  # 设置默认交易量
            df['market_type'] = '农贸市场'  # 设置默认市场类型
            
            # 确保数值列的数据类型
            numeric_columns = ['avg_price', 'low_price', 'high_price', 'data_quality_score']
            for col in numeric_columns:
                if col in df.columns:
                    df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0)
            
            # 处理日期列
            if 'pub_date' in df.columns:
                df['pub_date'] = pd.to_datetime(df['pub_date'], errors='coerce').dt.strftime('%Y-%m-%d')
            
            if 'crawl_time' in df.columns:
                df['crawl_time'] = pd.to_datetime(df['crawl_time'], errors='coerce').dt.strftime('%Y-%m-%d %H:%M:%S')
            
            logging.info("Hive真实数据清理完成")
            return df
            
        except Exception as e:
            logging.error(f"清理Hive真实数据失败: {e}")
            return df

    def load_sample_data(self):
        """加载示例数据（当Hive不可用时）"""
        try:
            np.random.seed(42)
            dates = pd.date_range('2024-12-01', periods=180, freq='D')
            
            # 包含原始名称和变体名称的产品列表
            all_products = []
            for standard_name, aliases in self.product_aliases.items():
                all_products.append(standard_name)
                all_products.extend(aliases)
            
            # 添加其他标准产品
            for product in self.standard_products:
                if product not in [name for aliases in self.product_aliases.values() for name in aliases]:
                    all_products.append(product)
            
            places = ['冀', '津鲁', '山东', '河北', '北京', '上海', '广州', '深圳', '杭州', '南京', 
                     '武汉', '成都', '西安', '重庆', '天津', '苏州', '青岛', '大连', '厦门', '宁波']
            
            data = []
            record_count = 0
            
            # 为每个产品设置基础价格
            base_prices = {
                '大白菜': 1.2, '小白菜': 2.5, '萝卜': 1.5, '土豆': 2.8, '番茄': 4.2, 
                '黄瓜': 3.5, '茄子': 3.8, '辣椒': 5.2, '豆角': 6.5, '苹果': 8.5,
                '梨': 6.8, '桃子': 9.2, '葡萄': 12.5, '香蕉': 5.8, '西瓜': 3.2,
                '胡萝卜': 2.8, '菠菜': 4.5, '韭菜': 3.8, '芹菜': 3.2, '生菜': 4.2,
                '菜花': 5.5, '豆芽': 2.2, '蒜苗': 6.8,
                # 变体名称使用相似价格
                '白萝卜': 1.5, '青萝卜': 1.8, '水萝卜': 2.2, '樱桃萝卜': 4.5,
                '青椒': 5.2, '红椒': 5.8, '尖椒': 6.2, '甜椒': 4.8, '干辣椒': 15.5,
                '西红柿': 4.2, '圣女果': 8.5, '马铃薯': 2.8, '洋芋': 2.8,
                '西芹': 3.2, '本芹': 3.0, '水芹': 3.5, '芹菜叶': 2.8
            }
            
            for date in dates:
                daily_products = np.random.choice(all_products, min(len(all_products), 20), replace=False)
                
                for product in daily_products:
                    daily_places = np.random.choice(places, np.random.randint(5, 10), replace=False)
                    
                    for place in daily_places:
                        standard_name = self.fuzzy_match_product(product)
                        base_price = base_prices.get(product, base_prices.get(standard_name, 3.0))
                        
                        season_factor = 1 + 0.2 * np.sin(2 * np.pi * date.dayofyear / 365)
                        region_factor = np.random.uniform(0.8, 1.2)
                        random_factor = 1 + np.random.normal(0, 0.1)
                        
                        avg_price = base_price * season_factor * region_factor * random_factor
                        low_price = avg_price * np.random.uniform(0.85, 0.95)
                        high_price = avg_price * np.random.uniform(1.05, 1.15)
                        
                        record = {
                            'record_id': f"{product}_{date.strftime('%Y%m%d')}_{place}_{record_count}",
                            'product_name': product,
                            'standard_product_name': standard_name,
                            'place': place,
                            'avg_price': round(max(0.3, avg_price), 2),
                            'low_price': round(max(0.2, low_price), 2),
                            'high_price': round(max(0.5, high_price), 2),
                            'unit': '斤',
                            'pub_date': date.strftime('%Y-%m-%d'),
                            'crawl_time': date.strftime('%Y-%m-%d %H:%M:%S'),
                            'price_anomaly_flag': 'NORMAL',
                            'data_quality_score': np.random.randint(85, 101),
                            'processing_timestamp': date.strftime('%Y-%m-%d %H:%M:%S'),
                            'volume': np.random.uniform(100, 2000),
                            'market_type': np.random.choice(['批发市场', '零售市场', '农贸市场', '超市'])
                        }
                        data.append(record)
                        record_count += 1

            self.df = pd.DataFrame(data)
            logging.info(f"生成示例数据成功，共 {len(self.df)} 条记录")
            return True

        except Exception as e:
            logging.error(f"生成示例数据失败: {e}")
            return False
    def manual_refresh_data(self):
        """手动刷新数据 - 重新连接Hive并加载最新数据"""
        try:
            refresh_info = {
                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'status': 'success',
                'message': '',
                'data_info': {}
            }
            
            logging.info("开始手动刷新数据...")
            
            # 1. 重新连接Hive
            old_conn_status = self.hive_conn is not None
            self.connect_hive()
            new_conn_status = self.hive_conn is not None
            
            # 2. 重新加载数据
            data_loaded = self.load_data()
            
            if data_loaded and self.df is not None:
                # 3. 收集数据信息
                refresh_info['data_info'] = {
                    'total_records': len(self.df),
                    'unique_products': self.df['product_name'].nunique(),
                    'unique_standard_products': self.df.get('standard_product_name', self.df['product_name']).nunique(),
                    'unique_places': self.df['place'].nunique(),
                    'date_range': {
                        'min': self.df['pub_date'].min(),
                        'max': self.df['pub_date'].max()
                    },
                    'data_source': 'Hive数据库' if new_conn_status else '模拟数据',
                    'hive_available': HIVE_AVAILABLE and new_conn_status
                }
                
                if new_conn_status and not old_conn_status:
                    refresh_info['message'] = f"成功连接到Hive并加载 {len(self.df)} 条记录"
                elif new_conn_status and old_conn_status:
                    refresh_info['message'] = f"Hive数据已同步，共 {len(self.df)} 条记录"
                else:
                    refresh_info['status'] = 'warning'
                    refresh_info['message'] = f"Hive连接失败，使用模拟数据 {len(self.df)} 条记录"
                    
                logging.info(f"手动刷新完成: {refresh_info['message']}")
                
            else:
                refresh_info['status'] = 'error'
                refresh_info['message'] = "数据加载失败"
                refresh_info['data_info'] = {
                    'total_records': 0,
                    'data_source': '无数据',
                    'hive_available': False
                }
                
            return refresh_info
            
        except Exception as e:
            logging.error(f"手动刷新数据失败: {e}")
            return {
                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'status': 'error',
                'message': f"刷新失败: {str(e)}",
                'data_info': {
                    'total_records': 0,
                    'data_source': '错误',
                    'hive_available': False
                }
            }

 def get_refresh_status(self):
        """获取当前刷新状态信息"""
        try:
            status_info = {
                'hive_available': HIVE_AVAILABLE,
                'hive_connected': self.hive_conn is not None,
                'data_source': 'Hive数据库' if self.hive_conn else '模拟数据',
                'total_records': len(self.df) if self.df is not None else 0,
                'last_refresh_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'data_loaded': self.df is not None and not self.df.empty
            }
            
            return status_info
            
        except Exception as e:
            logging.error(f"获取刷新状态失败: {e}")
            return {
                'hive_available': False,
                'hive_connected': False,
                'data_source': '未知',
                'total_records': 0,
                'last_refresh_time': None,
                'data_loaded': False
            }

    def force_reload_from_hive(self):
        """强制从Hive重新加载数据"""
        try:
            logging.info("强制从Hive重新加载数据...")
            
            # 关闭现有连接
            if self.hive_conn:
                try:
                    self.hive_conn.close()
                except:
                    pass
                self.hive_conn = None
            
            # 重新连接
            hive_connected = self.connect_hive()
            
            if hive_connected:
                # 尝试从Hive加载数据
                success = self.load_data_from_hive()
                if success:
                    logging.info(f"从Hive成功重新加载 {len(self.df)} 条记录")
                    return True
                else:
                    logging.warning("从Hive加载数据失败，回退到模拟数据")
                    return self.load_sample_data()
            else:
                logging.warning("Hive连接失败，使用模拟数据")
                return self.load_sample_data()
                
        except Exception as e:
            logging.error(f"强制重新加载失败: {e}")
            return self.load_sample_data()

    def search_products_in_df(self, df, search_term):
        """在DataFrame中搜索产品"""
        if not search_term:
            return df
        
        mask = (
            df['product_name'].str.contains(search_term, case=False, na=False) |
            df.get('standard_product_name', pd.Series(dtype='object')).str.contains(search_term, case=False, na=False) |
            df['place'].str.contains(search_term, case=False, na=False)
        )
        
        return df[mask]

    def get_overview_data(self, filters=None):
        """获取概览数据 - 修复版本，支持筛选"""
        try:
            # 应用筛选条件
            df = self.apply_filters(filters) if filters else self.df
            
            if df is None or df.empty:
                return {
                    'total_records': 0,
                    'unique_products': 0,
                    'unique_standard_products': 0,
                    'unique_places': 0,
                    'avg_price_today': 0,
                    'price_change': 0,
                    'last_update': '',
                    'hot_products': [],
                    'total_volume': 0
                }

            today = datetime.now().date()
            yesterday = today - timedelta(days=1)

            # 安全的日期转换
            try:
                today_data = df[pd.to_datetime(df['pub_date'], errors='coerce').dt.date == today]
                yesterday_data = df[pd.to_datetime(df['pub_date'], errors='coerce').dt.date == yesterday]
            except Exception as e:
                logging.error(f"日期处理错误: {e}")
                today_data = pd.DataFrame()
                yesterday_data = pd.DataFrame()

            # 安全的价格计算
            today_avg = today_data['avg_price'].mean() if not today_data.empty and 'avg_price' in today_data.columns else 0
            yesterday_avg = yesterday_data['avg_price'].mean() if not yesterday_data.empty and 'avg_price' in yesterday_data.columns else 0

            # 处理NaN值
            today_avg = 0 if pd.isna(today_avg) else today_avg
            yesterday_avg = 0 if pd.isna(yesterday_avg) else yesterday_avg

            price_change = round(today_avg - yesterday_avg, 2) if yesterday_avg > 0 else 0

            # 安全的热门产品计算
            try:
                if 'standard_product_name' in df.columns and 'volume' in df.columns:
                    hot_products = df.groupby('standard_product_name')['volume'].sum().nlargest(3).index.tolist()
                else:
                    hot_products = []
            except Exception as e:
                logging.error(f"热门产品计算错误: {e}")
                hot_products = []

            # 安全的总交易量计算
            try:
                total_volume = df['volume'].sum() if 'volume' in df.columns else 0
                total_volume = 0 if pd.isna(total_volume) else total_volume
            except Exception as e:
                logging.error(f"交易量计算错误: {e}")
                total_volume = 0

            # 转换所有数值为内置类型，避免jsonify序列化错误
            result = {
                'total_records': int(len(df)),
                'unique_products': int(df['product_name'].nunique()) if 'product_name' in df.columns else 0,
                'unique_standard_products': int(df.get('standard_product_name', df['product_name']).nunique()),
                'unique_places': int(df['place'].nunique()) if 'place' in df.columns else 0,
                'avg_price_today': float(round(today_avg, 2)),
                'price_change': float(price_change),
                'last_update': str(df['crawl_time'].max()) if 'crawl_time' in df.columns else '',
                'hot_products': hot_products,
                'total_volume': float(round(total_volume, 2))
            }

            return result

        except Exception as e:
            logging.error(f"获取概览数据失败: {e}")
            # 返回默认值而不是空字典
            return {
                'total_records': 0,
                'unique_products': 0,
                'unique_standard_products': 0,
                'unique_places': 0,
                'avg_price_today': 0,
                'price_change': 0,
                'last_update': '',
                'hot_products': [],
                'total_volume': 0
            }
    def get_price_comparison_data(self, filters=None):
        """获取价格对比数据"""
        try:
            df = self.apply_filters(filters) if filters else self.df
            
            if df.empty:
                return {}
            
            group_column = 'standard_product_name' if 'standard_product_name' in df.columns else 'product_name'
            
            price_stats = df.groupby(group_column).agg({
                'low_price': 'mean',
                'avg_price': 'mean', 
                'high_price': 'mean'
            }).round(2)
            
            return {
                'categories': price_stats.index.tolist(),
                'low_prices': price_stats['low_price'].tolist(),
                'avg_prices': price_stats['avg_price'].tolist(),
                'high_prices': price_stats['high_price'].tolist()
            }
        except Exception as e:
            logging.error(f"获取价格对比数据失败: {e}")
            return {}

    def get_price_trends_data(self, product_name, filters=None):
        """获取价格趋势数据"""
        try:
            df = self.apply_filters(filters) if filters else self.df
            
            if df.empty:
                return {}
            
            product_df = df[
                (df['product_name'] == product_name) |
                (df.get('standard_product_name', pd.Series(dtype='object')) == product_name)
            ].copy()
            
            if product_df.empty:
                return {}
            
            daily_trends = product_df.groupby('pub_date').agg({
                'low_price': 'mean',
                'avg_price': 'mean',
                'high_price': 'mean'
            }).round(2).reset_index()
            
            daily_trends = daily_trends.sort_values('pub_date')
            
            return {
                'dates': daily_trends['pub_date'].tolist(),
                'low_prices': daily_trends['low_price'].tolist(),
                'avg_prices': daily_trends['avg_price'].tolist(),
                'high_prices': daily_trends['high_price'].tolist()
            }
        except Exception as e:
            logging.error(f"获取价格趋势数据失败: {e}")
            return {}

    def get_place_analysis_data(self, filters=None):
        """获取产地分析数据"""
        try:
            df = self.apply_filters(filters) if filters else self.df
            
            if df.empty:
                return {}
            
            group_column = 'standard_product_name' if 'standard_product_name' in df.columns else 'product_name'
            
            place_stats = df.groupby([group_column, 'place']).agg({
                'avg_price': 'mean'
            }).round(2).reset_index()
            
            products = sorted(place_stats[group_column].unique().tolist())
            places = sorted([p for p in place_stats['place'].unique().tolist() if p])
            
            series_data = []
            colors = ['#5470c6', '#91cc75', '#fac858', '#ee6666', '#73c0de', '#3ba272', '#fc8452', '#9a60b4']
            
            for i, place in enumerate(places):
                place_data = place_stats[place_stats['place'] == place]
                prices = []
                for product in products:
                    product_price = place_data[place_data[group_column] == product]['avg_price']
                    prices.append(float(product_price.iloc[0]) if not product_price.empty else 0)
                
                series_data.append({
                    'name': place,
                    'type': 'bar',
                    'data': prices,
                    'itemStyle': {'color': colors[i % len(colors)]}
                })
            
            return {
                'categories': products,
                'series': series_data
            }
        except Exception as e:
            logging.error(f"获取产地分析数据失败: {e}")
            return {}
    def get_data_table(self, page=1, limit=5000, filters=None, search_term=None):
        """获取数据表格 - 修复版本"""
        try:
            # 先应用筛选条件
            df = self.apply_filters(filters) if filters else self.df
            
            # 再应用搜索条件
            if search_term:
                df = self.search_products_in_df(df, search_term)
            
            if df.empty:
                return {'total': 0, 'page': page, 'limit': limit, 'data': []}
            
            max_limit = 50000
            if limit > max_limit:
                limit = max_limit
            
            start_idx = (page - 1) * limit
            end_idx = start_idx + limit
            page_data = df.iloc[start_idx:end_idx]
            
            data_list = []
            for _, row in page_data.iterrows():
                data_list.append({
                    'product_name': row['product_name'],
                    'standard_product_name': row.get('standard_product_name', row['product_name']),
                    'low_price': float(row['low_price']),
                    'avg_price': float(row['avg_price']),
                    'high_price': float(row['high_price']),
                    'place': row['place'] if row['place'] else '未知',
                    'unit': row['unit'],
                    'pub_date': row['pub_date'],
                    'data_quality_score': int(row['data_quality_score']),
                    'volume': float(row.get('volume', 0)),
                    'market_type': row.get('market_type', '未知')
                })
            
            return {
                'total': int(len(df)),
                'page': int(page),
                'limit': int(limit),
                'data': data_list,
                'total_pages': (len(df) + limit - 1) // limit,
                'showing_count': len(data_list)
            }
        except Exception as e:
            logging.error(f"获取数据表格失败: {e}")
            return {'total': 0, 'page': page, 'limit': limit, 'data': []}

    def get_filters_options(self):
        """获取筛选器选项"""
        try:
            if self.df is None or self.df.empty:
                return {}
            
            standard_products = []
            if 'standard_product_name' in self.df.columns:
                standard_products = sorted(self.df['standard_product_name'].dropna().unique().tolist())
            
            original_products = sorted(self.df['product_name'].unique().tolist())
            
            return {
                'products': original_products,
                'standard_products': standard_products,
                'places': sorted([p for p in self.df['place'].unique().tolist() if p]),
                'market_types': sorted(self.df.get('market_type', pd.Series(dtype='object')).dropna().unique().tolist()),
                'date_range': {
                    'min': self.df['pub_date'].min(),
                    'max': self.df['pub_date'].max()
                }
            }
        except Exception as e:
            logging.error(f"获取筛选器选项失败: {e}")
            return {}

    def apply_filters(self, filters):
        """应用筛选条件 - 修复版本"""
        if not filters or self.df is None:
            return self.df
        
        df = self.df.copy()
        print(f"原始数据量: {len(df)}")  # 调试信息
        
        # 产品筛选 - 修复逻辑
        if 'products' in filters and filters['products']:
            print(f"应用产品筛选: {filters['products']}")
            
            # 创建筛选条件
            product_mask = pd.Series([False] * len(df))
            
            for selected_product in filters['products']:
                # 精确匹配原始产品名称
                mask1 = df['product_name'] == selected_product
                
                # 精确匹配标准产品名称
                if 'standard_product_name' in df.columns:
                    mask2 = df['standard_product_name'] == selected_product
                    product_mask = product_mask | mask1 | mask2
                else:
                    product_mask = product_mask | mask1
            
            df = df[product_mask]
            print(f"产品筛选后记录数: {len(df)}")
        
        # 产地筛选
        if 'places' in filters and filters['places']:
            print(f"应用产地筛选: {filters['places']}")
            df = df[df['place'].isin(filters['places'])]
            print(f"产地筛选后记录数: {len(df)}")
        
        # 市场类型筛选
        if 'market_types' in filters and filters['market_types']:
            if 'market_type' in df.columns:
                df = df[df['market_type'].isin(filters['market_types'])]
        
        # 日期筛选
        if 'start_date' in filters and filters['start_date']:
            df = df[pd.to_datetime(df['pub_date'], errors='coerce') >= pd.to_datetime(filters['start_date'])]
        
        if 'end_date' in filters and filters['end_date']:
            df = df[pd.to_datetime(df['pub_date'], errors='coerce') <= pd.to_datetime(filters['end_date'])]
        
        print(f"最终筛选结果: {len(df)} 条记录")
        return df
# Flask应用
app = Flask(__name__)
CORS(app)

# 初始化数据服务
data_service = AgricultureDataService()

@app.route('/')
def dashboard():
    """仪表板主页"""
    return render_template('dashboard.html')

@app.route('/api/overview')
def api_overview():
    """概览数据API - 修复版本"""
    try:
        filters = {
            'products': request.args.getlist('products'),
            'start_date': request.args.get('start_date'),
            'end_date': request.args.get('end_date'),
            'places': request.args.getlist('places'),
            'market_types': request.args.getlist('market_types')
        }
        
        # 清理空值
        filters = {k: v for k, v in filters.items() if v}
        
        data = data_service.get_overview_data(filters)
        return jsonify(data)
    except Exception as e:
        logging.error(f"概览API错误: {e}")
        return jsonify({'error': str(e)}), 500

@app.route('/api/price_comparison')
def api_price_comparison():
    """价格对比数据API"""
    try:
        filters = {
            'products': request.args.getlist('products'),
            'start_date': request.args.get('start_date'),
            'end_date': request.args.get('end_date'),
            'places': request.args.getlist('places'),
            'market_types': request.args.getlist('market_types')
        }
        
        filters = {k: v for k, v in filters.items() if v}
        
        data = data_service.get_price_comparison_data(filters)
        return jsonify(data)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/price_trends')
def api_price_trends():
    """价格趋势数据API"""
    try:
        product_name = request.args.get('product', '萝卜')
        
        filters = {
            'start_date': request.args.get('start_date'),
            'end_date': request.args.get('end_date'),
            'places': request.args.getlist('places')
        }
        
        filters = {k: v for k, v in filters.items() if v}
        
        data = data_service.get_price_trends_data(product_name, filters)
        return jsonify(data)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/place_analysis')
def api_place_analysis():
    """产地分析数据API"""
    try:
        filters = {
            'products': request.args.getlist('products'),
            'start_date': request.args.get('start_date'),
            'end_date': request.args.get('end_date'),
            'market_types': request.args.getlist('market_types')
        }
        
        filters = {k: v for k, v in filters.items() if v}
        
        data = data_service.get_place_analysis_data(filters)
        return jsonify(data)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/data_table')
def api_data_table():
    """数据表格API - 修复版本"""
    try:
        page = int(request.args.get('page', 1))
        limit = int(request.args.get('limit', 5000))
        search_term = request.args.get('search', '').strip()
        export_format = request.args.get('export', '')
        
        if request.args.get('show_all') == 'true':
            limit = 100000
        
        filters = {
            'products': request.args.getlist('products'),
            'start_date': request.args.get('start_date'),
            'end_date': request.args.get('end_date'),
            'places': request.args.getlist('places'),
            'market_types': request.args.getlist('market_types')
        }
        
        # 清理空值
        filters = {k: v for k, v in filters.items() if v}
        
        print(f"API接收到的筛选参数: {filters}")  # 调试信息
        
        if export_format == 'csv':
            df = data_service.apply_filters(filters) if filters else data_service.df
            if search_term:
                df = data_service.search_products_in_df(df, search_term)
            
            if not df.empty:
                output = io.StringIO()
                df.to_csv(output, index=False, encoding='utf-8-sig')
                output.seek(0)
                
                return Response(
                    output.getvalue(),
                    mimetype='text/csv',
                    headers={'Content-Disposition': 'attachment; filename=agriculture_data.csv'}
                )
            else:
                return jsonify({'error': '没有数据可导出'}), 400
        
        data = data_service.get_data_table(page, limit, filters, search_term)
        return jsonify(data)
    except Exception as e:
        logging.error(f"数据表格API错误: {e}")
        return jsonify({'error': str(e)}), 500

@app.route('/api/search')
def api_search():
    """产品搜索API"""
    try:
        search_term = request.args.get('q', '').strip()
        limit = int(request.args.get('limit', 1000))
        
        if not search_term:
            return jsonify({'results': []})
        
        df = data_service.df
        results = data_service.search_products_in_df(df, search_term)
        
        if not results.empty:
            summary = {
                'total_found': len(results),
                'products': results['product_name'].unique().tolist()[:10],
                'standard_products': results.get('standard_product_name', pd.Series(dtype='object')).dropna().unique().tolist()[:10],
                'avg_price_range': {
                    'min': float(results['avg_price'].min()),
                    'max': float(results['avg_price'].max()),
                    'mean': float(results['avg_price'].mean())
                }
            }
        else:
            summary = {'total_found': 0, 'products': [], 'standard_products': []}
        
        return jsonify({'results': summary})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/filters')
def api_filters():
    """获取筛选器选项"""
    try:
        data = data_service.get_filters_options()
        print(f"筛选器选项: {len(data.get('products', []))} 个产品, {len(data.get('places', []))} 个产地")
        return jsonify(data)
    except Exception as e:
        logging.error(f"获取筛选器选项失败: {e}")
        return jsonify({'error': str(e)}), 500

@app.route('/api/refresh_data', methods=['POST'])
def api_refresh_data():
    """手动刷新数据API"""
    try:
        logging.info("收到手动刷新请求")
        
        # 执行手动刷新
        refresh_result = data_service.manual_refresh_data()
        
        return jsonify(refresh_result)
        
    except Exception as e:
        logging.error(f"手动刷新API错误: {e}")
        return jsonify({
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'status': 'error',
            'message': f'刷新失败: {str(e)}',
            'data_info': {
                'total_records': 0,
                'data_source': '错误',
                'hive_available': False
            }
        }), 500

@app.route('/api/refresh_status')
def api_refresh_status():
    """获取刷新状态API"""
    try:
        status = data_service.get_refresh_status()
        return jsonify(status)
    except Exception as e:
        logging.error(f"获取刷新状态失败: {e}")
        return jsonify({'error': str(e)}), 500

@app.route('/api/force_hive_reload', methods=['POST'])
def api_force_hive_reload():
    """强制从Hive重新加载数据"""
    try:
        logging.info("收到强制Hive重载请求")
        
        success = data_service.force_reload_from_hive()
        
        if success:
            data_info = {
                'total_records': len(data_service.df) if data_service.df is not None else 0,
                'data_source': 'Hive数据库' if data_service.hive_conn else '模拟数据',
                'hive_available': data_service.hive_conn is not None
            }
            
            return jsonify({
                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'status': 'success',
                'message': f"强制重载成功，共 {data_info['total_records']} 条记录",
                'data_info': data_info
            })
        else:
            return jsonify({
                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'status': 'error',
                'message': '强制重载失败',
                'data_info': {
                    'total_records': 0,
                    'data_source': '失败',
                    'hive_available': False
                }
            }), 500
            
    except Exception as e:
        logging.error(f"强制Hive重载API错误: {e}")
        return jsonify({
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'status': 'error',
            'message': f'强制重载失败: {str(e)}',
            'data_info': {
                'total_records': 0,
                'data_source': '错误',
                'hive_available': False
            }
        }), 500

if __name__ == '__main__':
    print("🌱 农业数据可视化平台启动中...")
    print("📊 支持的农产品类型:")
    for i, product in enumerate(data_service.standard_products, 1):
        print(f"   {i:2d}. {product}")
    print(f"\n🔍 支持模糊搜索和别名匹配")
    print(f"📈 支持大数据量显示（最多100,000条记录）")
    print(f"🔄 支持手动刷新Hive数据同步")
    print(f"🌐 访问地址: http://localhost:5000")
    
    if HIVE_AVAILABLE and data_service.hive_conn:
        print("✅ 已连接到Hive数据库")
    else:
        print("⚠️  使用模拟数据（Hive不可用）")
    
    print("=" * 50)
    
    app.run(host='0.0.0.0', port=5000, debug=True, threaded=True)


